package com.github.rpc4j.common.network.handler;

import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;

import org.apache.log4j.Logger;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelUpstreamHandler;

import com.github.rpc4j.common.ServiceRequest;
import com.github.rpc4j.common.ServiceResponse;
import com.github.rpc4j.common.exception.ServiceCallException;

public class NettyRpc4jClientHandler extends SimpleChannelUpstreamHandler {
    
	private final static Logger logger = Logger.getLogger(NettyRpc4jClientHandler.class);
	
	public ConcurrentHashMap<Integer, CountDownLatch> requestLatchMap = new ConcurrentHashMap<Integer, CountDownLatch>();
	public ConcurrentHashMap<Integer, ServiceResponse> responseMap = new ConcurrentHashMap<Integer, ServiceResponse>();
	
	@Override
	public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
	}
	

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e)
			throws Exception {
		ServiceResponse response = new ServiceResponse();
		Throwable throwable = e.getCause();
		response.setThrowable(throwable);
		System.out.println("error: "+throwable.getMessage());
		System.out.println("error put ChannelId:"+ctx.getChannel().getId());
		responseMap.put(ctx.getChannel().getId(), response);
		CountDownLatch latch = requestLatchMap.remove(ctx.getChannel().getId());
		latch.countDown();
	}
	

	@Override
	public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
			throws Exception {
		ServiceResponse response =  (ServiceResponse)e.getMessage();
		System.out.println("put ChannelId:"+response.getRequestChannelId());
		responseMap.put(response.getRequestChannelId(), response);
		CountDownLatch latch = requestLatchMap.remove(response.getRequestChannelId());
		latch.countDown();
	}
	
	
	public ServiceResponse serviceCall(ServiceRequest request,Channel channel) throws ServiceCallException, InterruptedException, ExecutionException{
		   CountDownLatch latch = new CountDownLatch(1);
		   requestLatchMap.put(request.getChannelId(), latch);
		   Request requestCallable = new Request(request,latch,channel);
		   FutureTask<ServiceResponse>  futureTask = new FutureTask<ServiceResponse>(requestCallable);
		   futureTask.run();
		   return futureTask.get();
	}
	
	
	public class Request implements Callable<ServiceResponse>{
		private ServiceRequest request;
		private CountDownLatch latch = null;
		private Channel channel;
		public Request(ServiceRequest request,CountDownLatch latch,Channel channel){
			this.request = request;
			this.latch = latch;
			this.channel = channel;
		}

		@Override
		public ServiceResponse call() throws Exception {
			channel.write(request);
			latch.await();
			System.out.println("get  ChannelId:"+request.getChannelId());
			ServiceResponse response = responseMap.remove(request.getChannelId());
			return response;
		}
		
	}
	
    
}
